package com.fourth.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fourth.bean.TrafficHomeDetailPageViewBean;
import com.fourth.utils.DateFormatUtil;
import com.fourth.utils.MyClickHouseUtil;
import com.fourth.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author wjy
 * @create 2022-08-20 9:43
 */
public class DwsTrafficPageViewWindow {
    public static void main(String[] args) throws Exception {
        //1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取kafka dwd层页面日志主题数据创建流---为什么不直接用uv主题明细数据，可能会丢数据（同一mid用户可能既访问了首页又访问了商品详情页，那么后续做去重后只能得到第一条，第二条就丢了）
        String topic = "dwd_traffic_page_log";
        String groupId = "page_view_uv_22";
        DataStreamSource<String> pageViewDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //3.将数据转为JSON对象并过滤出页面为首页和商品详情页的数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = pageViewDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                if ("home".equals(pageId) || "good_detail".equals(pageId)) {
                    out.collect(jsonObject);
                }
            }
        });

        //4.提取事件时间并生成watermark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWmDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));


        //5.按照mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWmDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //6.使用状态编程对分组过后的数据做去重，当天首次登录的数据才记录，并转为JavaBean
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> homeDetailPageViewDS = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, TrafficHomeDetailPageViewBean>() {

            private ValueState<String> homeState;
            private ValueState<String> detailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义状态描述器
                ValueStateDescriptor<String> homeDesc = new ValueStateDescriptor<String>("home-state", String.class);
                ValueStateDescriptor<String> detailDesc = new ValueStateDescriptor<String>("detail-state", String.class);

                //设置ttl 状态保存一天 创建和写时更新(重置)状态
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();

                homeDesc.enableTimeToLive(ttlConfig);
                detailDesc.enableTimeToLive(ttlConfig);

                homeState = getRuntimeContext().getState(homeDesc);
                detailState = getRuntimeContext().getState(detailDesc);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                //获取状态里的日期以及当前数据中的page_id和日期
                String homeLastDt = homeState.value();
                String detailLastDt = detailState.value();
                String pageId = value.getJSONObject("page").getString("page_id");
                String curDt = DateFormatUtil.toDate(value.getLong("ts"));

                //定义首页即商品详情页的访客人数，初始化为0L，也方便后续操作
                long homeUvCt = 0L;
                long detailUvCt = 0L;
                //去重逻辑 判断是首页还是商品详情页 状态中日期等于null 状态中的末次日期不等于今天
                if ("home".equals(pageId) && (homeLastDt == null || !homeLastDt.equals(curDt))) {
                    //将首页访客人数记为1
                    homeUvCt = 1L;
                    //更新状态
                    homeState.update(curDt);
                } else if ("good_detail".equals(pageId) && (detailLastDt == null || !detailLastDt.equals(curDt))) {
                    detailUvCt = 1L;
                    detailState.update(curDt);
                }

                //写出数据
                if (homeUvCt == 1L || detailUvCt == 1L) {
                    out.collect(new TrafficHomeDetailPageViewBean("","",homeUvCt,detailUvCt,null));
                }
            }
        });

        //7.开窗（非键控流开窗只能用WindowAll）、聚合（还要补全字段）
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduceDS = homeDetailPageViewDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
            @Override
            public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {

                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());
                return value1;
            }
        }, new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<TrafficHomeDetailPageViewBean> values, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                //获取数据
                TrafficHomeDetailPageViewBean next = values.iterator().next();
                //补充字段
                next.setTs(System.currentTimeMillis());
                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                //输出数据
                out.collect(next);
            }
        });

        //8.写出到ClickHouse
        reduceDS.print("reduceDS>>>>>>>");
        reduceDS.addSink(MyClickHouseUtil.getSink("insert into dws_traffic_page_view_window values(?,?,?,?,?)"));

        //9.启动任务
        env.execute("DwsTrafficPageViewWindow");
    }
}
